package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.StockSecListPutWindowFunction;
import cn.itcast.function.StockSecWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.sink.SinkHbase;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 个股秒级行情
 * 1.每5s生成一条最新的数据
 */
public class StockSecTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {

        /**
         * 开发梳理:
         * 1.每只股票每5s会采集5条数据
         * 2.秒级行情是每5s生成一条最新的数据
         * 3.需要获取最新时间的数据,每只股票的最新时间的一条数据
         * 4.根据股票唯一标识做分组操作,划分时间窗口(5s)(会形成一个个的窗口,每个窗口内会保存同一个股票代码的数据)
         * 5.根据时间获取最新时间的数据,需要比较时间大小
         */

        /**
         * 开发步骤：
         * 1.数据分组
         * 2.划分时间窗口
         * 3.新建个股数据写入bean对象
         * 4.秒级窗口函数业务处理
         * 5.数据写入操作
         *   * 封装ListPuts
         *   * 数据写入
         */
        //1.数据分组
        waterData.keyBy(new KeyFunction())
                // 2.划分时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //4.秒级窗口函数业务处理
                .apply(new StockSecWindowFunction())
                //5.数据写入操作
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new StockSecListPutWindowFunction())
                .addSink(new SinkHbase(QuotConfig.config.getProperty("stock.hbase.table.name")));



    }
}
